package com.ly.kafka.service.impl;

import com.ly.kafka.service.KafkaService;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

@Service
public class KafkaServiceImpl implements KafkaService {


    @Autowired
    private AdminClient kafkaAdminClient;


    @Override
    public KafkaFuture<Map<String, TopicDescription>> SelectTopicInfo(String topicName) {
        DescribeTopicsResult result = kafkaAdminClient.describeTopics(Arrays.asList(topicName));
        KafkaFuture<Map<String, TopicDescription>> all = result.all();
        return all;
    }

    @Override
    public void editTopicPartitionNum(String topicName, Integer number) {
        Map<String, NewPartitions> newPartitions=new HashMap<String, NewPartitions>();
        //创建新的分区的结果
        newPartitions.put(topicName,NewPartitions.increaseTo(number));
        kafkaAdminClient.createPartitions(newPartitions);
    }
}
